package com.yjxxt;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * 创建生产者线程类
 */
public class Hello01Producer extends Thread {
    //创建Kafka的生产者
    private Producer<String, String> producer;

    /**
     * 创建构造器
     */
    public Hello01Producer(String producerName) {
        //设置线程的名字
        super.setName(producerName);
        //创建配置文件列表
        Properties properties = new Properties();
        // kafka地址，多个地址用逗号分割
        properties.put("bootstrap.servers", "192.168.255.11:9092,192.168.255.12:9092,192.168.255.13:9092");
        //设置写出数据的格式
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        //写出的应答方式
        properties.put("acks", "all");
        //错误重试
        properties.put("retries", 1);
        //批量写出
        properties.put("batch.size", 16384);
        //创建生产者对象
        producer = new KafkaProducer<String, String>(properties);
    }

    @Override
    public void run() {
        //初始化一个计数器
        int count = 0;

        System.out.println("Hello01Producer.run--开始发送数据");
        //迭代發送消息
        while (count < 100) {
            count++;
            String key = "yjx01-" + count;
            String value = Thread.currentThread().getName() + "--" + count;
            //封装消息对象
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("userlog", key, value);
            //发送消息到服务器
            producer.send(producerRecord);
            //打印消息
            //System.out.println("Producer.run--" + key + "--" + value);
            //每个1秒发送1条
            //try {
            //    Thread.sleep(1000);
            //} catch (InterruptedException e) {
            //    e.printStackTrace();
            //}
        }
        //强行将数据写出
        producer.close();
        System.out.println("Hello01Producer.run--发送数据完毕");
    }

    public static void main(String[] args) {
        Hello01Producer producer = new Hello01Producer("优极限01");
        producer.start();
    }
}